/**
 *
 */
package com.rrd.hive.udaf;

import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StandardMapObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.io.Text;

/**
 * @author liubaoxin
 * @version 创建时间：2017年11月13日 上午10:44:43
 *          根据传入的参数依次为：行业、账号类型、周期、分数
 */
public class GroupToTongDun extends AbstractGenericUDAFResolver {
    private static String[] periodArr = new String[] { "7天内", "1个月内", "3个月内",
        "6个月内", "12个月内", "18个月内", "24个月内", "近60个月以上" };

    private static String[] industryArr = new String[] { "网上银行", "信用卡中心",
        "银行消费金融公司", "银行对公业务", "银行个人业务", "银行小微贷款", "直销银行", "财产保险", "人身保险",
        "大型消费金融公司", "小额贷款公司", "P2P网贷", "厂商汽车金融", "一般消费分期平台", "汽车金融", "房地产金融",
        "大数据金融", "融资租赁", "担保", "资产转让", "交通工具租赁", "设备租赁", "房屋租赁", "汽车租赁",
        "互联网金融门户", "第三方服务商", "众筹", "理财机构", "综合类电商平台", "比价类电商平台", "跨境电商平台",
        "团购平台", "垂直类电商平台", "信息中介" };
    private static String[] accountArr = new String[] { "id_number",
        "account_mobile", "account_email", "device_id" };

    @Override
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)
            throws SemanticException {
        if (parameters.length >= 4) {
            return new FilterConcatEvaluator();

        } else {
            throw new UDFArgumentTypeException(parameters.length, "输入参数不正确!");
        }

    }

    public static class FilterConcatEvaluator extends GenericUDAFEvaluator {
        private PrimitiveObjectInspector inputOI; // 行业
        private PrimitiveObjectInspector inputOI2;  //账号类型
        private PrimitiveObjectInspector inputOI3;  //周期
        private PrimitiveObjectInspector inputOI4;  //分数
        private StandardMapObjectInspector mapOI;
        private Text result = new Text();

        @Override
        public ObjectInspector init(Mode m, ObjectInspector[] parameters)
                throws HiveException {
            super.init(m, parameters);

            if (m == Mode.PARTIAL1) {
                this.inputOI = (PrimitiveObjectInspector) parameters[0];
                this.inputOI2 = (PrimitiveObjectInspector) parameters[1];
                this.inputOI3 = (PrimitiveObjectInspector) parameters[2];
                this.inputOI4 = (PrimitiveObjectInspector) parameters[3];
                return ObjectInspectorFactory.getStandardMapObjectInspector(
                    ObjectInspectorUtils.getStandardObjectInspector(
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector),
                    PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.PARTIAL2) {
                this.mapOI = (StandardMapObjectInspector) parameters[0];
                return ObjectInspectorFactory.getStandardMapObjectInspector(
                    PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                    PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.FINAL) {
                this.mapOI = (StandardMapObjectInspector) parameters[0];
//                return ObjectInspectorFactory
//                        .getStandardMapObjectInspector(
//                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
//                                PrimitiveObjectInspectorFactory.javaStringObjectInspector);
                return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
            } else if (m == Mode.COMPLETE) {
//            	return ObjectInspectorFactory
//                        .getStandardMapObjectInspector(
//                                PrimitiveObjectInspectorFactory.javaStringObjectInspector,
//                                PrimitiveObjectInspectorFactory.javaStringObjectInspector);
                return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
            } else { //
                throw new RuntimeException("no such mode Exception");
            }
        }

        static class ResultAgg implements AggregationBuffer {
            Map<String, String> map = new HashMap<>();
        }

        @Override
        public AggregationBuffer getNewAggregationBuffer()
                throws HiveException {
            // TODO Auto-generated method stub
            AggregationBuffer aggr = new ResultAgg();
            this.reset(aggr);
            return aggr;
        }

        @Override
        public void reset(AggregationBuffer agg) throws HiveException {
            // TODO Auto-generated method stub
            ((ResultAgg) agg).map = new HashMap<>();

        }

        @Override
        public void iterate(AggregationBuffer agg, Object[] parameters)
                throws HiveException {
            // TODO Auto-generated method stub
            if (parameters == null || parameters[0] == null) {
                return;
            }

            ResultAgg rag = (ResultAgg) agg;

            String key = PrimitiveObjectInspectorUtils.getString(parameters[0],
                    this.inputOI)
                    + PrimitiveObjectInspectorUtils.getString(parameters[1],
                    this.inputOI2)
                    + PrimitiveObjectInspectorUtils.getString(parameters[2],
                    this.inputOI3);
            rag.map.put(key, PrimitiveObjectInspectorUtils
                    .getString(parameters[3], this.inputOI4));
        }

        @Override
        public Object terminatePartial(AggregationBuffer agg)
                throws HiveException {
            // TODO Auto-generated method stub
            ResultAgg myagg = (ResultAgg) agg;
            Map<String, String> ret = new HashMap<>(myagg.map);

            return ret;
        }

        @Override
        public Object terminate(AggregationBuffer agg) throws HiveException {
            // TODO Auto-generated method stub
            ResultAgg aggr = (ResultAgg) agg;
            if (aggr.map == null || aggr.map.size() == 0) {
                // System.out.println("result is null");
                return null;
            }
            this.result.set(this.matrix(aggr.map).toString());
            return this.result;
        }

        private String matrix(Map<String, String> map) {

            StringBuilder sb = new StringBuilder();
            for (String s : GroupToTongDun.industryArr) {
                for (String s2 : GroupToTongDun.accountArr) {
                    for (String s3 : GroupToTongDun.periodArr) {

                        sb.append(map.get(s + s2 + s3) == null ? "0"
                            : map.get(s + s2 + s3)).append(",");
                    }
                }
            }
            //行业对应分数
            for (String s : GroupToTongDun.industryArr) {
                for (String s3 : GroupToTongDun.periodArr) {
                    sb.append(map.get(s + "ALL" + s3) == null ? "0"
                        : map.get(s + "ALL" + s3)).append(",");
                }
            }
            //账号类型对应分数
            for (String s2 : GroupToTongDun.accountArr) {
                for (String s3 : GroupToTongDun.periodArr) {
                    sb.append(map.get("ALL" + s2 + s3) == null ? "0"
                        : map.get("ALL" + s2 + s3)).append(",");
                }
            }
            for (String s3 : GroupToTongDun.periodArr) {
                sb.append(map.get("ALL" + "ALL" + s3) == null ? "0"
                    : map.get("ALL" + "ALL" + s3)).append(",");
            }
            sb.delete(sb.length() - 1, sb.length());

            return sb.toString();

        }

        @Override
        public void merge(AggregationBuffer agg, Object partial)
                throws HiveException {
            // TODO Auto-generated method stub
            ResultAgg aggr = (ResultAgg) agg;
            if (partial != null) { // 在没有输入数据情况下，null会传进来
                Map partialResult = this.mapOI.getMap(partial);
                for (Object key : partialResult.keySet()) {
                    aggr.map.put(key.toString(),
                        partialResult.get(key).toString());

                }
            }
        }
    }

}
